package com.zhu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhu.bean.TradePaymentWindowBean;
import com.zhu.config.KafkaTopicConfig;
import com.zhu.utils.ClickHouseUtil;
import com.zhu.utils.DataFormatUtil;
import com.zhu.utils.TimestampLtz3CompareUtil;
import com.zhu.utils.ZhuKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 *DWS 层交易成功窗口汇总表
 *
 * 涉及到左链接 存在重复数据  按照单一字段分组 保留最晚到来的数据 利用状态变成 设置存活时间和上游left join 流表在内存中的存活时间一直 5s
 *
 * 运用 Flink 状态编程，在状态中维护用户末次支付日期。
 * 若末次支付日期为 null，则将首次支付用户数和支付独立用户数均置为 1；否则首次
 * 支付用户数置为 0，判断末次支付日期是否为当日，如果不是当日则支付独立用户数置为 1，
 * 否则置为 0。最后将状态中的支付日期更新为当日。
 */
public class DWSTradePaymentSucWindowApp {

    public static void main(String[] args) throws Exception {
        //todo env
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        streamExecutionEnvironment.setParallelism(4);
        //checkPoint
        /*
        streamExecutionEnvironment.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE); //精确一次

        //状态后端
        streamExecutionEnvironment.setStateBackend(new HashMapStateBackend());
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointStorage(ClusterParametersConfig.HDFS_CHECKPOINT_FILE_DIR);  //检查点保存在hdfs
        System.setProperty("HADOOP_USER_NAME", "zhu");
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);  //TimeOut
        streamExecutionEnvironment.getCheckpointConfig().setMaxConcurrentCheckpoints(2);  //最大共存检查点
        streamExecutionEnvironment.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5 * 1000L));  //重启策略

         */
        //todo kafka topic dwd_trade_pay_detail_suc
        String topic = KafkaTopicConfig.KAFKA_DWD_TRADE_PAYMENT_SUC_TOPIC;
        String groupId = "dws_trade_payment_suc_window" + KafkaTopicConfig.KAFKA_GROUP_ID_LAST_NAME;
        DataStreamSource<String> kafkaDStream = streamExecutionEnvironment.addSource(ZhuKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        //todo json
        SingleOutputStreamOperator<JSONObject> jsonObjDStream = kafkaDStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> collector) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    collector.collect(jsonObject);
                } catch (Exception e) {
                    System.out.println("Dirty>>>" + value);
                    e.printStackTrace();
                }
            }
        });

        //todo 订单明细id 分组
        KeyedStream<JSONObject, String> jsonObjKeyByDetailIdDStream
                = jsonObjDStream.keyBy(jsonObject -> jsonObject.getString("order_detail_id"));

        jsonObjKeyByDetailIdDStream.print("jsonDATA");

        //todo 去重
        SingleOutputStreamOperator<JSONObject> filterDStream = jsonObjKeyByDetailIdDStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {

            private ValueState<JSONObject> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                ValueStateDescriptor<JSONObject> valueStateDescriptor = new ValueStateDescriptor<JSONObject>("value-state", JSONObject.class);
                valueState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public void processElement(JSONObject value, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context,
                                       Collector<JSONObject> collector) throws Exception {

                JSONObject state = valueState.value();
                if (state == null) {
                    valueState.update(value);
                    context.timerService().registerProcessingTimeTimer(context.timerService().currentProcessingTime() + 5000L);
                } else {
                    String startRt = state.getString("row_op_ts");
                    String currentRt = value.getString("row_op_ts");

                    int compare = TimestampLtz3CompareUtil.compare(startRt, currentRt);  //前面大返回1
                    if (compare != 1) {
                        valueState.update(value);
                    }
                }
            }

            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                super.onTimer(timestamp, ctx, out);
                //输青并清空状态
                JSONObject value = valueState.value();
                out.collect(value);
                valueState.clear();
            }
        });


        //todo waterMark
        SingleOutputStreamOperator<JSONObject> jsonObjWithWaterMarkDStream = filterDStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObject, long l) {
                        return DataFormatUtil.toTs(jsonObject.getString("callback_time"), true);
                    }
                }));

        //todo user_id 分组
        KeyedStream<JSONObject, String> keyedByUserDStream = jsonObjWithWaterMarkDStream.keyBy(jsonObject -> jsonObject.getString("user_id"));

        //todo state 处理数据格式 判断是否是 独立支付 或着 首次支付用户
        SingleOutputStreamOperator<TradePaymentWindowBean> tradePaymentDStream = keyedByUserDStream.flatMap(
                new RichFlatMapFunction<JSONObject, TradePaymentWindowBean>() {

                    private ValueState<String> lastDtState;

                    @Override
                    public void flatMap(JSONObject data, Collector<TradePaymentWindowBean> collector) throws Exception {

                        String lastDt = lastDtState.value();
                        String dataDt = data.getString("callback_time").split(" ")[0];

                        //当日独立用户  &  新增支付用户
                        Long paymentSucUniqueUserCount = 0L;
                        Long paymentSucNewUserCount = 0L;
                        if (lastDt == null) {
                            paymentSucNewUserCount = 1L;
                            paymentSucUniqueUserCount = 1L;
                            lastDtState.update(dataDt);
                        } else if (!lastDt.equals(dataDt)) {
                            paymentSucUniqueUserCount = 1L;
                            lastDtState.update(dataDt);
                        }
                        if (paymentSucUniqueUserCount == 1L) {
                            collector.collect(new TradePaymentWindowBean("", "", paymentSucUniqueUserCount, paymentSucNewUserCount, null));
                        }
                    }

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        ValueStateDescriptor<String> lastDateStateDescriptor = new ValueStateDescriptor<>("last_date", String.class);
                        lastDtState = getRuntimeContext().getState(lastDateStateDescriptor);
                    }
                }

        );

        //todo window
        AllWindowedStream<TradePaymentWindowBean, TimeWindow> windowAllTradePaymentDStream = tradePaymentDStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)));

        //todo reduce
        SingleOutputStreamOperator<TradePaymentWindowBean> resultDStream = windowAllTradePaymentDStream.reduce(
                new ReduceFunction<TradePaymentWindowBean>() {
                    @Override
                    public TradePaymentWindowBean reduce(TradePaymentWindowBean value1, TradePaymentWindowBean value2) throws Exception {
                        value1.setPaymentSucNewUserCount(value1.getPaymentSucNewUserCount() + value2.getPaymentSucNewUserCount());
                        value1.setPaymentSucUniqueUserCount(value1.getPaymentSucUniqueUserCount() + value2.getPaymentSucUniqueUserCount());
                        return value1;
                    }
                },
                new AllWindowFunction<TradePaymentWindowBean, TradePaymentWindowBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow timeWindow, Iterable<TradePaymentWindowBean> iterable, Collector<TradePaymentWindowBean> collector) throws Exception {
                        String start = DataFormatUtil.toYmdHms(timeWindow.getStart());
                        String end = DataFormatUtil.toYmdHms(timeWindow.getEnd());
                        long ts = System.currentTimeMillis();
                        TradePaymentWindowBean bean = iterable.iterator().next();
                        bean.setTs(ts);
                        bean.setStt(start);
                        bean.setEdt(end);
                        collector.collect(bean);
                    }
                }
        );
        //todo write clickhouse
        resultDStream.print(">>>");
        resultDStream.addSink(ClickHouseUtil.getClickHouseSinkFunction("" +
                "insert into dws_trade_payment_suc_window values(?,?,?,?,?))"));



        //todo execute job
        streamExecutionEnvironment.execute("DWSTradePaymentSucWindowApp");

    }
}
